package net.bw.realtime.tmall.dws.funcion1;

import net.bw.realtime.tmall.dwd.bean.TransactionEventBean;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import net.bw.realtime.tmall.common.utils.JdbcUtil;

import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * @Author: liuyawei
 * @Date: 2025/5/28
 */

public class ToDeviceDisWindowFunction {
	public static void toDeviceDistribution(DataStream<TransactionEventBean> timeStream)
	{
		SingleOutputStreamOperator<String> device_distribution = timeStream
				.filter(x -> "pay_success".equals(x.getActionType()))
				.keyBy(x -> x.getDeviceType())
				.window(TumblingEventTimeWindows.of(Time.seconds(10)))
				.aggregate(new AggregateFunction<TransactionEventBean, Tuple2<String, Double>, Tuple2<String, Double>>() {
					@Override
					public Tuple2<String, Double> createAccumulator() {
						return Tuple2.of("", 0.0);
					}

					@Override
					public Tuple2<String, Double> add(TransactionEventBean TransactionEventBean, Tuple2<String, Double> stringDoubleTuple2) {
						Double amount = TransactionEventBean.getQuantity() * TransactionEventBean.getPrice();
						return Tuple2.of(TransactionEventBean.getDeviceType(), amount + stringDoubleTuple2.f1);
					}

					@Override
					public Tuple2<String, Double> getResult(Tuple2<String, Double> stringDoubleTuple2) {
						return stringDoubleTuple2;
					}

					@Override
					public Tuple2<String, Double> merge(Tuple2<String, Double> stringDoubleTuple2, Tuple2<String, Double> acc1) {
						return Tuple2.of(stringDoubleTuple2.f0, stringDoubleTuple2.f1 + acc1.f1);
					}
				}, new ProcessWindowFunction<Tuple2<String, Double>, String, String, TimeWindow>() {
					private final SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

					@Override
					public void process(String s, ProcessWindowFunction<Tuple2<String, Double>, String, String, TimeWindow>.Context context, Iterable<Tuple2<String, Double>> iterable, Collector<String> collector) throws Exception {
						Tuple2<String, Double> next = iterable.iterator().next();
						String window_start = sdf.format(new Date(context.window().getStart()));
						String window_end = sdf.format(new Date(context.window().getEnd()));

						collector.collect(window_start + "," + window_end + "," + next.f0 + "," + next.f1 + "," + System.currentTimeMillis());
					}

				}).name("device-distribution-aggregation")
				.uid("device-distribution-aggregation");

		//device_distribution.print();

		JdbcUtil.sinkToClickhouseUpsert(device_distribution,
				"INSERT INTO jtp_tmall_report.dws_device_distribution_report(\n" +
						"    window_start_time, window_end_time, device_type, total_amount, ts\n" +
						") VALUES (\n" +
						"    ?, ?, ?, ?, ?\n" +
						")"
		);


	}


}
